#include <boost/algorithm/string.hpp>

#include "spdlog/spdlog.h"
#include "rtmp_server_session.hpp"
#include "base/network/socket_interface.hpp"

#include "core/session.hpp"
#include "core/source_manager.hpp"
#include "core/rtmp_media_source.hpp"
#include "core/rtmp_media_sink.hpp"

#include "server/rtmp/amf0/amf0_inc.hpp"

#include "server/rtmp/rtmp_message/chunk_message/rtmp_acknowledge_message.hpp"

#include "server/rtmp/rtmp_message/command_message/rtmp_connect_command_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_connect_resp_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_set_peer_bandwidth_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_window_ack_size_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_release_stream_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_release_stream_resp_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_create_stream_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_create_stream_resp_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_fcpublish_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_fcpublish_resp_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_publish_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_play_message.hpp"
#include "server/rtmp/rtmp_message/command_message/user_ctrl_message/stream_begin_message.hpp"
#include "server/rtmp/rtmp_message/command_message/rtmp_onstatus_message.hpp"

using namespace mms;

RtmpServerSession::RtmpServerSession(std::shared_ptr<SocketInterface> conn):
    StreamSession(conn->get_worker()),conn_(conn),handshkae_(conn),chunk_protocol_(conn),
    recv_coroutine_exited_(get_worker()->get_io_context()),
    send_coroutine_exited_(get_worker()->get_io_context()),
    rtmp_msgs_channel_(get_worker()->get_io_context(),1024)
{

    set_session_type("rtmp");
}

RtmpServerSession::~RtmpServerSession() {
    spdlog::debug("destroy RtmpServerSession");
}

void RtmpServerSession::service(){
    start_recv_coroutine();
}

void RtmpServerSession::start_recv_coroutine(){
    auto self(shared_from_this());
    recv_coroutine_running_ = true;
    boost::asio::co_spawn(conn_->get_worker()->get_io_context(), [this,self]()
        ->boost::asio::awaitable<void> {
            // 启动握手
        spdlog::info("  // 启动握手");
        if(!co_await handshkae_.do_server_handshake()){
                co_return;
        }
            // chunk处理
        spdlog::info("RtmpSerrverSession do_server_handshake ok");
        // 启动发送数据的协程
        spdlog::info(" 启动发送数据的协程");
        start_send_coroutine();
        spdlog::info(" RtmpSetChunkSizeMessage");
        chunk_protocol_.set_out_chunk_size(409600);
        RtmpSetChunkSizeMessage set_chunk_size_msg(chunk_protocol_.get_out_chunk_size());

        // 发送setchunk 消息
        if (!co_await send_rtmp_message(set_chunk_size_msg))
        {
            co_return;
        }
        int32_t ret = co_await chunk_protocol_.cycle_recv_rtmp_message(std::bind(&RtmpServerSession::on_recv_rtmp_message,this,std::placeholders::_1));
        co_return;
        
    },[this,self](std::exception_ptr exp){
        (void)exp;
        close();
        recv_coroutine_exited_.close();
        recv_coroutine_running_ = false;
        spdlog::info("RtmpServerSession recv coroutine exited");
    });
}

boost::asio::awaitable<void> RtmpServerSession::stop_recv_coroutine()
{
    if (recv_coroutine_running_)
    {
        if (recv_coroutine_exited_.is_open())
        {
            boost::system::error_code ec;
            co_await recv_coroutine_exited_.async_receive(boost::asio::redirect_error(boost::asio::use_awaitable, ec));
        }
    }
    co_return;
}

void RtmpServerSession::start_send_coroutine()
{
    // 启动发送协程
    send_coroutine_running_ = true;
    auto self(shared_from_this());
    boost::asio::co_spawn(conn_->get_worker()->get_io_context(),[this,self]()
        ->boost::asio::awaitable<void>{
            boost::system::error_code ec;
            while(1){
                 auto [rtmp_msgs, ch] = co_await rtmp_msgs_channel_.async_receive(boost::asio::redirect_error(boost::asio::use_awaitable, ec));
                if(ec){
                    break;
                }
                if(!(co_await chunk_protocol_.send_rtmp_messages(rtmp_msgs))){
                    if(ch){
                        ch->close();
                    }
                    co_return;
                }
                if(ch){
                    ch->close();
                }
            }
        },[this,self](std::exception_ptr exp){
            (void)exp;
            close();
            send_coroutine_exited_.close();
            send_coroutine_running_ = false;
            spdlog::debug("RtmpServerSession send coroutine exited");
        });
}

boost::asio::awaitable<void> RtmpServerSession::stop_send_coroutine(){
    if(send_coroutine_running_){
        if(rtmp_msgs_channel_.is_open()){
            rtmp_msgs_channel_.close();
        }
        if(send_coroutine_exited_.is_open()){
            boost::system::error_code ec;
            co_await send_coroutine_exited_.async_receive(boost::asio::redirect_error(boost::asio::use_awaitable,ec));
        }
    }
    spdlog::info("RtmpServerSession send coroutine exited");
    co_return;
}

// 异步发送Rtmpmesage
boost::asio::awaitable<bool> RtmpServerSession::send_rtmp_message(const 
    std::vector<std::shared_ptr<RtmpMessage>>& rtmp_msgs){
        co_await rtmp_msgs_channel_.async_send(boost::system::error_code{},
            rtmp_msgs, nullptr, boost::asio::use_awaitable);
        co_return true;
    }


boost::asio::awaitable<bool> mms::RtmpServerSession::on_recv_rtmp_message(std::shared_ptr<RtmpMessage> &rtmp_msg)
{
    // spdlog::info("RtmpServerSession::on_recv_rtmp_message ::got a rtmp message, type: {}", rtmp_msg->message_type_id_);
    if(! co_await send_acknowledge_msg_if_required()){
        co_return false;
    }

    if (chunk_protocol_.is_protocol_control_message(rtmp_msg))
    {
        if(!co_await chunk_protocol_.handle_protocol_control_message(rtmp_msg)){
            co_return -18;
        }
        co_return true;
    }

    switch (rtmp_msg->get_message_type())
    {
        case RTMP_MESSAGE_TYPE_AMF0_COMMAND:
        {
            co_return co_await handle_amf0_command(rtmp_msg);
        }
        case RTMP_MESSAGE_TYPE_AMF0_DATA:
        {
            co_return handle_amf0_data(rtmp_msg);
        }
        case RTMP_MESSAGE_TYPE_AUDIO:
        {
            co_return handle_audio_msg(rtmp_msg);
        }
        case RTMP_MESSAGE_TYPE_VIDEO:
        {
            co_return handle_video_msg(rtmp_msg);
        }
    }
    co_return true;
}


boost::asio::awaitable<bool> RtmpServerSession::send_acknowledge_msg_if_required(){
    int64_t delat = chunk_protocol_.get_last_ack_bytes() - conn_->get_in_bytes();
    if(delat >= chunk_protocol_.get_in_window_acknowledge_size() / 2){
        RtmpAcknwledgeMessage ack_msg(conn_->get_in_bytes());
        if(!co_await send_rtmp_message(ack_msg)){
            co_return false;
        }
        chunk_protocol_.set_last_ack_bytes(conn_->get_in_bytes());
        co_return true;
    }
    co_return true;
}


boost::asio::awaitable<bool> RtmpServerSession::handle_amf0_command(std::shared_ptr<RtmpMessage> rtmp_msg){
    Amf0String command_name;
    int32_t consumed = command_name.decode(rtmp_msg->payload_,rtmp_msg->payload_size_);
    if(consumed < 0){
        spdlog::error("decode rtmp command msg failed, code:{}", consumed);
        co_return false;
    }



    auto name = command_name.get_value();
    spdlog::info("server get amf0 name: {}",name);
    if(name == RTMP_COMMAND_NAME_CONNECT){
        co_return co_await handle_amf0_connect_command(rtmp_msg);
    }
    else if (name == "releaseStream")
    {
        co_return co_await handle_amf0_release_stream_command(rtmp_msg);
    }
    else if (name == "FCPublish")
    {
        co_return co_await handle_amf0_fcpublish_command(rtmp_msg);
    }
    else if (name == "createStream")
    {
        co_return co_await handle_amf0_createstream_command(rtmp_msg);
    }
    else if (name == "publish")
    {
        co_return co_await handle_amf0_publish_command(rtmp_msg);
    }else if(name == "play"){
        co_return co_await handle_amf0_play_command(rtmp_msg);
    }
    co_return true;
}

boost::asio::awaitable<bool> RtmpServerSession::handle_amf0_connect_command(std::shared_ptr<RtmpMessage> rtmp_msg){
    RtmpConnectCommandMessage connect_command;
    auto consumed = connect_command.decode(rtmp_msg);
    if(consumed < 0){
        co_return false;
    }
    // 获取domain app stream_name
    if (!parse_connect_cmd(connect_command))
    {
        co_return false;
    }
    RtmpWindowAckSizeMessage window_ack_size_msg(chunk_protocol_.get_in_window_acknowledge_size());
    if(!co_await send_rtmp_message(window_ack_size_msg)){
        co_return false;
    }
    RtmpSetPeerBandWidthMessage set_peer_bandwidth_msg(800000000, LIMIT_TYPE_DYNAMIC);
    if (!co_await send_rtmp_message(set_peer_bandwidth_msg))
    {
        co_return false;
    }

    RtmpConnectRespMessage result_msg(connect_command, "_result");
    result_msg.props().set_item_value("fmsVer", "FMS/3,0,1,123");
    result_msg.props().set_item_value("capabilities", 31);

    result_msg.info().set_item_value("level", "status");
    result_msg.info().set_item_value("code", RTMP_RESULT_CONNECT_SUCCESS);
    result_msg.info().set_item_value("description", "Connection succeed.");
    if (!co_await send_rtmp_message(result_msg))
    {
        co_return false;
    }
    co_return true;

}

bool RtmpServerSession::parse_connect_cmd(RtmpConnectCommandMessage &connect_command)
{
    { // parse domain
        std::vector<std::string> vs;
        boost::split(vs, connect_command.tc_url_, boost::is_any_of("/"));
        if (vs.size() < 3)
        {
            return false;
        }

        if (vs[0] != "rtmp:" && vs[0] != "rtmps:")
        {
            return false;
        }

        if (vs[1] != "")
        {
            return false;
        }

        domain_ = vs[2];
        if (domain_.find(":") != std::string::npos)
        { // 去掉端口号
            std::vector<std::string> tmp;
            boost::split(tmp, domain_, boost::is_any_of(":"));
            if (tmp.size() > 1)
            {
                domain_ = tmp[0];
            }
        }
    }

    { // parse app, stream, params
        std::vector<std::string> vs;
        boost::split(vs, connect_command.app_, boost::is_any_of("/"));
        if (vs.size() < 1)
        {
            return false;
        }

        app_name_ = vs[0];
        if (vs.size() >= 2)
        { // 兼容obs推流时，可能将流名写到前面
            // todo 考虑带参数的情况
            stream_name_ = vs[1];
            auto qmark_pos = stream_name_.find("?");
            if (qmark_pos != std::string::npos)
            {
                stream_name_ = stream_name_.substr(0, qmark_pos);
            }
        }
    }

    {
        auto question_mark_pos = connect_command.tc_url_.find("?");
        if (question_mark_pos != std::string::npos)
        { // 解析错误
            std::vector<std::string> vs;
            std::string params_list = connect_command.tc_url_.substr(question_mark_pos + 1, connect_command.tc_url_.size() - question_mark_pos - 1);
            boost::split(vs, params_list, boost::is_any_of("&"));
            for (auto &s : vs)
            {
                auto equ_pos = s.find("=");
                if (equ_pos == std::string::npos)
                {
                    continue;
                }

                std::string name = s.substr(0, equ_pos);
                std::string value = s.substr(equ_pos + 1);
                set_param(name, value);
            }
        }
    }
    return true;
}

boost::asio::awaitable<bool> RtmpServerSession::handle_amf0_release_stream_command(std::shared_ptr<RtmpMessage> rtmp_msg)
{
    RtmpReleaseStreamMessage release_command;
    auto consumed = release_command.decode(rtmp_msg);
    if (consumed < 0)
    {
        co_return false;
    }

    RtmpReleaseStreamRespMessage result_msg(release_command, "_result");
    if (!co_await send_rtmp_message(result_msg))
    {
        co_return false;
    }
    co_return true;
}

boost::asio::awaitable<bool> RtmpServerSession::handle_amf0_fcpublish_command(std::shared_ptr<RtmpMessage> rtmp_msg)
{
    RtmpFCPublishMessage fcpublish_cmd;
    auto consumed = fcpublish_cmd.decode(rtmp_msg);
    if (consumed < 0)
    {
        co_return false;
    }

    RtmpFCPublishRespMessage result_msg(fcpublish_cmd, "_result");
    if (!co_await send_rtmp_message(result_msg))
    {
        co_return false;
    }
    co_return true;
}

boost::asio::awaitable<bool> RtmpServerSession::handle_amf0_createstream_command(std::shared_ptr<RtmpMessage> rtmp_msg)
{
    RtmpCreateStreamMessage create_stream_cmd;
    auto consumed = create_stream_cmd.decode(rtmp_msg);
    if (consumed < 0)
    {
        spdlog::error("handle_amf0_createstream_command decoe failed");
        co_return false;
    }

    spdlog::info("get create stream command");
    RtmpCreateStreamRespMessage result_msg(create_stream_cmd, "_result");
    if (!co_await send_rtmp_message(result_msg))
    {
        spdlog::info("handle_amf0_createstream_command failed");
        co_return false;
    }
    co_return true;
}

bool RtmpServerSession::parse_publish_cmd(RtmpPublishMessage &pub_cmd)
{
    auto &stream_name = pub_cmd.stream_name();
    if (stream_name_.empty())
    {
        stream_name_ = stream_name;
    }
    set_session_info(domain_, app_name_, stream_name_);
    spdlog::info("set session info, domain:{}, app:{}, stream:{}", domain_, app_name_, stream_name_);
    return true;
}

bool RtmpServerSession::parse_play_cmd(RtmpPlayMessage &play_cmd)
{
    auto &stream_name = play_cmd.stream_name();
    if (stream_name_.empty())
    {
        stream_name_ = stream_name;
    }

    session_name_ = domain_ + "/" + app_name_ + "/" + stream_name_;
    set_session_info(domain_, app_name_, stream_name_);
    return true;
}

boost::asio::awaitable<bool> RtmpServerSession::handle_amf0_publish_command(std::shared_ptr<RtmpMessage> rtmp_msg)
{
    RtmpPublishMessage publish_cmd;
    auto consumed = publish_cmd.decode(rtmp_msg);
    if (consumed < 0)
    {
        co_return false;
    }

    if (!parse_publish_cmd(publish_cmd))
    {
        co_return false;
    }

    // 1. 查找已有source
    is_publisher_ = true;
    auto self(std::static_pointer_cast<RtmpServerSession>(shared_from_this()));
    auto rtmp_media_source = SourceManager::get_instance().get_source(get_session_name());
    if (!rtmp_media_source)
    { // 如果不存在，则创建
        rtmp_media_source_ = std::make_shared<RtmpMediaSource>(get_worker(), std::weak_ptr<StreamSession>(self));
        rtmp_media_source_->set_session(self);
        rtmp_media_source_->set_source_info(domain_, app_name_, stream_name_);
    }
    else
    {
        // 如果存在，并且绑定旧的session，那么关闭旧的session
        auto old_session = std::static_pointer_cast<RtmpServerSession>(rtmp_media_source->get_session());
        if (old_session)
        {
            old_session->detach_source();
            old_session->close(); // 关闭老的推流端
        }
        // 绑定新的session
        rtmp_media_source->set_session(self);
        rtmp_media_source_ = std::static_pointer_cast<RtmpMediaSource>(rtmp_media_source);
    }

    if (!rtmp_media_source_)
    {
        spdlog::error("create rtmp media source failed");
        co_return false;
    }

    RtmpOnStatusMessage status_msg;
    status_msg.info().set_item_value("level", "status");
    status_msg.info().set_item_value("code", RTMP_STATUS_STREAM_PUBLISH_START);
    status_msg.info().set_item_value("description", "publish start ok.");
    status_msg.info().set_item_value("clientid", "mms");
    if (!co_await send_rtmp_message(status_msg))
    {
        spdlog::error("send publish resp failed");
        co_return false;
    }

    spdlog::info("publish process succeed");
    SourceManager::get_instance().add_source_if_not_exist(session_name_, rtmp_media_source_);
    co_return true;
}

boost::asio::awaitable<bool> RtmpServerSession::handle_amf0_play_command(std::shared_ptr<RtmpMessage> rtmp_msg)
{
    RtmpPlayMessage play_cmd;
    auto consumed = play_cmd.decode(rtmp_msg);
    if (consumed < 0)
    {
        spdlog::error("parse play cmd failed, consumed:{}", consumed);
        co_return false;
    }
    // 1. 解析cmd信息
    if (!parse_play_cmd(play_cmd))
    {
        spdlog::error("parse play cmd failed");
        co_return false;
    }

    auto self(std::static_pointer_cast<RtmpServerSession>(shared_from_this()));
    // 2. 查找源流
    // todo: how to record 404 error to log.
    auto source = SourceManager::get_instance().get_source(get_session_name());
    if (!source)
    { // todo:发送STREAM NOT FOUND
        spdlog::error("could not find source:{}", get_session_name());
        // 4. 发送onStatus信息
        RtmpOnStatusMessage status_msg;
        status_msg.info().set_item_value("level", "status");
        status_msg.info().set_item_value("code", RTMP_STATUS_STREAM_NOT_FOUND);
        status_msg.info().set_item_value("description", "play start ok.");
        status_msg.info().set_item_value("clientid", "mms");
        if (!co_await send_rtmp_message(status_msg))
        {
            spdlog::error("send on status for play cmd failed.");
            co_return false;
        }
        co_return false;
    }

    // 3. 发送播放开始消息
    spdlog::info("发送播放开始消息");
    RtmpStreamBeginMessage stream_begin_msg(1); // 只用1这个stream_id
    if (!co_await send_rtmp_message(stream_begin_msg))
    {
        spdlog::error("send stream begin message failed.");
        co_return false;
    }
    // 4. 发送onStatus信息
    spdlog::info("发送onStatus信息");
    RtmpOnStatusMessage status_msg;
    status_msg.info().set_item_value("level", "status");
    status_msg.info().set_item_value("code", RTMP_STATUS_STREAM_PLAY_START);
    status_msg.info().set_item_value("description", "play start ok.");
    status_msg.info().set_item_value("clientid", "mms");
    if (!co_await send_rtmp_message(status_msg))
    {
        spdlog::error("send on status for play cmd failed.");
        co_return false;
    }

    is_publisher_ = false;
    is_player_ = true;
    // 5. 设置sink的处理函数
    rtmp_media_sink_ = std::make_shared<RtmpMediaSink>(get_worker());
    rtmp_media_sink_->on_rtmp_message([this, self](const std::vector<std::shared_ptr<RtmpMessage>> &rtmp_msgs) -> boost::asio::awaitable<bool>
                                      { co_return co_await this->send_rtmp_message(rtmp_msgs); });

    rtmp_media_sink_->on_close([this, self]()
                               { this->close(); });

    bool ret = true;
    // 6. 添加到source的sink列表中
    ret = source->add_media_sink(rtmp_media_sink_);
    spdlog::info("add rtmp sink succeed");
    co_return ret;
}

bool RtmpServerSession::handle_video_msg(std::shared_ptr<RtmpMessage> msg)
{
    if (!rtmp_media_source_)
    {
        return false;
    }
    return rtmp_media_source_->on_video_packet(msg);
}

bool RtmpServerSession::handle_audio_msg(std::shared_ptr<RtmpMessage> msg)
{
    if (!rtmp_media_source_)
    {
        return false;
    }
    return rtmp_media_source_->on_audio_packet(msg);
}

bool RtmpServerSession::handle_amf0_data(std::shared_ptr<RtmpMessage> rtmp_msg)
{ // usually is metadata
    Amf0String command_name;
    int32_t consumed = command_name.decode(rtmp_msg->payload_, rtmp_msg->payload_size_);
    if (consumed < 0)
    {
        return false;
    }

    auto name = command_name.get_value();
    if (name == "onMetaData" || name == "@setDataFrame")
    {
        if (!rtmp_media_source_)
        {
            return false;
        }
        return rtmp_media_source_->on_metadata(rtmp_msg);
    }
    return true;
}

void RtmpServerSession::close()
{
    if(closed_.test_and_set(std::memory_order_acquire)){
        return;
    }
    auto self(shared_from_this());
    boost::asio::co_spawn(get_worker()->get_io_context(),[this,self]()
        ->boost::asio::awaitable<void> {
            boost::system::error_code ec;
            // auto sock = std::static_pointer_cast<TcpSocket>(conn_);
            // boost::asio::ip::tcp::endpoint remote_endpoint = sock->get_sock().remote_endpoint();
            std::string addr_str;
            unsigned short port;
            if(conn_){
                addr_str = conn_->addr_;
                port = conn_->port_;
                conn_->close();
            }
            co_await stop_recv_coroutine();
            co_await stop_send_coroutine();

            if (rtmp_media_source_)
            {
                SourceManager::get_instance().remove_source(get_session_name());
                rtmp_media_source_ = nullptr;
            }

            if (rtmp_media_sink_)
            {
                rtmp_media_sink_->close();
                rtmp_media_sink_->on_rtmp_message({});
                rtmp_media_sink_->on_close({});
                rtmp_media_sink_ = nullptr;
            }
            if(conn_){
                conn_.reset();
            }
       
            
            if(is_player_){
                spdlog::info("RtmpServerSession rtmp player closed");
            }
            else{
                 spdlog::info("RtmpServerSession rtmp pusher closed");
            }
            spdlog::info("addr: {}, port: {}", addr_str,port);
            co_return;
        }, boost::asio::detached);

 
}

void RtmpServerSession::detach_source()
{
    rtmp_media_source_ = nullptr;
}
